package com.yun.storm.pv;

import java.util.HashMap;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * 获取PVSpout发送的数据，PVTopology开启多线程。 给出每个线程处理的PV数
 *
 * 在多线程情况下，对PV数据只能局部汇总，不能整体汇总，可以把局部汇总的结果给一个单线程的BOLT进行整体汇总（PVSumBolt）
 *
 * @author WangXin
 * @version V1.0
 */
public class PVBolt extends BaseRichBolt {

	/**
	 *
	 */
	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	private Map<String, Integer> _counts = new HashMap<String, Integer>();

	/**
	 * 实例初始化的时候调用一次
	 *
	 * @param stormConf
	 *            The Storm configuration for this bolt.
	 * @param context
	 *            This object can be used to get information about this task's
	 *            place within the topology, including the task id and component
	 *            id of this task, input and output information, etc.
	 * @param collector
	 *            The collector is used to emit tuples from this bolt
	 */
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	/**
	 * Process a single tuple of input.
	 *
	 * @param input
	 *            The input tuple to be processed.
	 */
	@Override
	public void execute(Tuple input) {

		try {
//			String line = input.getString(0);
			byte[] binaryByField = input.getBinaryByField("bytes");
	        String line = new String(binaryByField);

			String reg = "/resource.\\S{0,}";
			// 计算每个网址被访问次数
			String[] URL = line.split("\t");
			System.out.println("日志分割后长度：######"+URL.length);
			System.out.println("日志分割内容"+URL);
			String url = "";
			String dateAccess="";
			boolean insertFlg = false;
			int count;
			if (URL != null && URL.length > 4) {
				if ("200".equals(URL[4])) {// 只统计状态码为200的日志
					if (URL[3].indexOf("?") != -1) {
						url = URL[3].substring(0, URL[3].indexOf("?")).replace("/jshop", "").replace("http://202.123.72.168:80", "").split(" ")[1];
					} else {
						url = URL[3].replace("/jshop", "").replace("http://202.123.72.168:80", "").split(" ")[1];
					}
					if (!url.matches(reg) && !"/".equals(url) && !"/ValidateImage.do".equals(url)) {// 筛除静态资源文件
						System.out.println(url);
						if (_counts.containsKey(url)) {
							count = _counts.get(url);
						} else {
							count = 0;
							insertFlg = true;
						}
						count++;
						_counts.put(url, count);
						dateAccess = URL[2].replace("[", "");
						dateAccess = dateAccess.substring(7, 11) + "-" + DateFormat(dateAccess.substring(3, 6))+"-"+dateAccess.substring(0, 2)+" "+dateAccess.substring(12, 20);
						collector.emit(new Values(url, count, dateAccess, insertFlg));
						collector.ack(input);
					}
				}
			}

		} catch (Exception e) {
			e.printStackTrace();
			this.collector.fail(input);
		}
	}

	/**
	 * Declare the output schema for all the streams of this topology.
	 *
	 * @param declarer
	 *            this is used to declare output stream ids, output fields, and
	 *            whether or not each output stream is a direct stream
	 */
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("URl", "pv", "date", "insertFlg"));
	}
	
	public static String DateFormat(String str1) {
		
		
		if ("Jan".equals(str1)) {
			return "01";
		} else if ("Feb".equals(str1)) {
			return "02";
		} else if ("Mar".equals(str1)) {
			return "03";
		} else if ("Apr".equals(str1)) {
			return "04";
		} else if ("May".equals(str1)) {
			return "05";
		} else if ("Jun".equals(str1)) {
			return "06";
		} else if ("Jul".equals(str1)) {
			return "07";
		} else if ("Aug".equals(str1)) {
			return "08";
		} else if ("Sep".equals(str1)) {
			return "09";
		} else if ("Oct".equals(str1)) {
			return "10";
		} else if ("Nov".equals(str1)) {
			return "11";
		} else if ("Dec".equals(str1)) {
			return "12";
		} else {
			return "";
		}
	}
}